package com.yujiahao.bigdata.acc

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/**
TODO 累加器，使用特殊的数据结构，告诉Spark，要将数据拉取回到Driver
      1、因为RDD 不能将和RDD无关的数据从Executor端拉取到Driver端
      2、每个Task计算时，都是传递累加器变量，没有shuffle的概念
      3、然后，计算完毕后，将累加器的结果返回到Driver端进行聚合（合并）
        1）声明累加器对象
        2）累加数据
        3）获取累加器累加的结果（只能在driver和读取和读各自的）
 */
object Spark01_Acc {
  def main(args: Array[String]): Unit = {
    //TODO 1、获得Spark环境的连接
    // 参数解读：setAppName("当前执行程序的名字").setMaster("当前的环境")
    // 分布式环境的基本架构为 主 - 从
    val conf: SparkConf = new SparkConf().setAppName("WordCount").setMaster("local")
    val sc: SparkContext = new SparkContext(conf)
    //TODO 3、业务逻辑
    val rdd: RDD[(Int)] = sc.makeRDD(List(1, 2, 3, 4), 2)

    //TODO 1、声明累加器对象
    var sum = sc.longAccumulator("sum")
    //sc.collectionAccumulator()
    //sc.doubleAccumulator()
    rdd.foreach(

      num => {
        //TODO 2、累加数据
        sum.add(num)
      }
    )
    //TODO 3、获取累加器累加的结果（只能在driver和读取和读各自的）
    println(sum.value)


  }
}
